package net.bwie.realtime.warehouse.DWD;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.DWD
 * @Author: zhangleqing
 * @CreateTime: 2025-09-01  17:09
 * @Description: TODO dwd_retreat_behavior （退后行为汇总表）
 * 相关表：order_base（订单基础信息表） refund_detail（退款明细信息表） consumer_behavior（消费者行为表） user（消费者信息表）
 * @Version: 1.0
 */
public class DwdRetreatBehavior {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1 开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2 读取数据
        readTable(tableEnv);

        // 3 数据清洗+加工
        Table resultTable = handle(tableEnv);

//        resultTable.execute().print();

        // 4 映射表创建
        createView(tableEnv);

        // 5 输出数据
        saveToSink(tableEnv, resultTable);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("DwdRetreatBehavior", resultTable);
        tableEnv.executeSql(
                "insert into dwd_retreat_behavior " +
                        " select * from DwdRetreatBehavior"
        ).print();
    }

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_retreat_behavior (\n" +
                        "    order_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    \n" +
                        "    refund_id STRING,\n" +
                        "    apply_time TIMESTAMP(3),\n" +
                        "    refund_scene STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    \n" +
                        "    behavior_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    behavior_type STRING,\n" +
                        "    PRIMARY KEY (order_id) NOT ENFORCED\n" +
                        ") WITH (\n" +
//                        "     'connector' = 'kafka',\n" +
//                        "     'topic' = 'dwd_retreat_behavior',\n" +
//                        "     'properties.bootstrap.servers' = 'node101:9092',\n" +
//                        "     'format' = 'json'\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dwd_retreat_behavior',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );
    }


/*
SELECT
    o.order_id,                -- 订单ID
    o.product_id,              -- 商品ID
    o.pay_time,                -- 订单支付时间
    r.refund_id,               -- 退款单号
    r.apply_time,              -- 退款申请时间
    r.refund_scene,            -- 退款场景
    r.refund_amount,           -- 实际退款金额
    cb.behavior_id,            -- 消费者行为ID
    o.user_id,                 -- 消费者ID（统一取订单表的user_id，确保一致性）
    cb.behavior_type           -- 行为类型（搜索/浏览/下单等）
FROM ods_order_base o  -- 核心表：订单表
LEFT JOIN ods_refund_detail r
    ON o.order_id = r.order_id  -- 订单ID匹配
    AND r.apply_time >= o.pay_time  -- 业务逻辑：退款申请时间≥支付时间（先支付后退款）
LEFT JOIN ods_consumer_behavior cb
    ON o.user_id = cb.user_id  -- 用户ID匹配（确保是订单归属用户的行为）
    AND cb.behavior_time BETWEEN o.pay_time - INTERVAL '1' DAY AND o.pay_time + INTERVAL '7' DAY  -- 行为时间范围：订单支付前后1天和7天（实际为支付前1天和支付后7天）
INNER JOIN ods_user u
    ON o.user_id = u.user_id  -- 确保user_id有效（无用户信息则过滤）
WHERE o.pay_time IS NOT NULL  -- 过滤订单表自身脏数据（支付时间为空的无效订单）


 */

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                "SELECT \n" +
                        "    o.order_id,                -- 订单ID\n" +
                        "    o.product_id,              -- 商品ID\n" +
                        "    o.pay_time,                -- 订单支付时间\n" +
                        "    r.refund_id,               -- 退款单号\n" +
                        "    r.apply_time,              -- 退款申请时间\n" +
                        "    r.refund_scene,            -- 退款场景\n" +
                        "    r.refund_amount,           -- 实际退款金额\n" +
                        "    cb.behavior_id,            -- 消费者行为ID\n" +
                        "    o.user_id,                 -- 消费者ID（统一取订单表的user_id，确保一致性）\n" +
                        "    cb.behavior_type           -- 行为类型（搜索/浏览/下单等）\n" +
                        "FROM ods_order_base o  -- 核心表：订单表\n" +
                        "LEFT JOIN ods_refund_detail r \n" +
                        "    ON o.order_id = r.order_id  -- 订单ID匹配\n" +
                        "    AND r.apply_time >= o.pay_time  -- 业务逻辑：退款申请时间≥支付时间（先支付后退款）\n" +
                        "LEFT JOIN ods_consumer_behavior cb \n" +
                        "    ON o.user_id = cb.user_id  -- 用户ID匹配（确保是订单归属用户的行为）\n" +
//                        "    AND cb.behavior_time BETWEEN o.pay_time - INTERVAL '1' DAY AND o.pay_time + INTERVAL '7' DAY  -- 行为时间范围：订单支付前后1天和7天（实际为支付前1天和支付后7天）\n" +
                        "LEFT JOIN ods_user FOR SYSTEM_TIME AS OF o.proctime AS u \n" +
                        "    ON o.user_id = u.user_id  -- 确保user_id有效（无用户信息则过滤）\n" +
                        // 过滤条件之所以这么严格是因为我们做到是退款后的统计
                        "WHERE o.user_id IS NOT NULL and o.pay_time is not null and r.refund_id is not null -- 过滤订单表自身脏数据（支付时间为空的无效订单）"
        );
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        // 1 order_base（订单基础信息表）
        tableEnv.executeSql(
                "create table ods_order_base (\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    shop_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "    click_ship_time TIMESTAMP(3),\n" +
                        "    order_amount DECIMAL(10,2),\n" +
                        "    order_status STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-order-base-Log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-order-base-Log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 2 refund_detail（退款明细信息表）
        tableEnv.executeSql(
                "create table ods_refund_detail (\n" +
                        "    refund_detail_id STRING,\n" +
                        "    refund_id STRING,\n" +
                        "    order_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    apply_time TIMESTAMP(3),\n" +
                        "    refund_status STRING,\n" +
                        "    finish_time TIMESTAMP(3),\n" +
                        "    product_id STRING,\n" +
                        "    refund_amount DOUBLE,\n" +
                        "    user_fill_reason STRING,\n" +
                        "    refund_scene STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR apply_time AS apply_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-refund-detail-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-refund-detail-log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 3 consumer_behavior（消费者行为表）
        tableEnv.executeSql(
                "create table ods_consumer_behavior (\n" +
                        "    behavior_id STRING,\n" +
                        "    user_id STRING,\n" +
                        "    behavior_type STRING,\n" +
                        "    behavior_time TIMESTAMP(3),\n" +
                        "    behavior_scenario STRING,\n" +
                        "    proctime AS PROCTIME(),\n" +
                        "    WATERMARK FOR behavior_time AS behavior_time - INTERVAL '5' SECOND\n" +
                        ") WITH (\n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods-consumer-behavior-log',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods-consumer-behavior-log',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );
        // 4 user（消费者信息表）
        tableEnv.executeSql(
                "CREATE TABLE ods_user (\n" +
                        "    user_id STRING,\n" +
                        "    user_name STRING,\n" +
                        "    user_age INT,\n" +
                        "    user_sex STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        // 新增：Doris FE 的 JDBC 端口（默认9030）
                        "    'jdbc-url' = 'jdbc:mysql://node102:9030',\n" +
                        "    'table.identifier' = 'transactions_ods.user',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        // 维表查询失败重试3次
                        "    'lookup.max-retries' = '3',\n"+
                        "    -- 移除不支持的'read.mode'参数（当前版本默认批处理模式）\n" +
                        "    'lookup.cache.max-rows' = '10000',  -- 支持的缓存参数\n" +
                        "    'lookup.cache.ttl' = '3600000'      -- 支持的缓存过期时间\n" +
                        ")"
        );
    }


    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
